package com.hq.tomato.ms.handler;

import com.hq.tomato.ms.config.TomatoMsProperties;
import com.hq.tomato.ms.entity.RetryMessage;
import com.hq.tomato.ms.util.JacksonUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.Map;

import static com.hq.tomato.ms.constants.RetryMessageConstants.CAN_CONTINUE;
import static com.hq.tomato.ms.constants.RetryMessageConstants.RETRY_NUM;

/**
 *
 *
 * @Description: <br> 重新发送消息的定时任务
 * @Project: <br>
 * @CreateDate: Created in 2019-11-23 14:44 <br>
 * @Author: <a href="1026365768@qq.com">qiang.hu</a>
 */
@Slf4j
@EnableConfigurationProperties(TomatoMsProperties.class)
public class ReSendMessageHandler{
    @Autowired
    private MongoTemplate mongoTemplate;
    @Autowired
    private TomatoMsProperties tomatoMsProperties;
    @Autowired
    private AmqpTemplate amqpTemplate;

    private final Integer DEFAULT_LIMIT = 100;



    public void execute() throws Exception {
        Integer retryNum = tomatoMsProperties.getMaxRetryNum();
        Query query = new Query();
        query.limit(DEFAULT_LIMIT);
        Criteria criteria = Criteria.where(CAN_CONTINUE).is(Boolean.TRUE).and(RETRY_NUM).lt(retryNum);
        query.addCriteria(criteria);
        List<RetryMessage> retryMessages = mongoTemplate.find(query, RetryMessage.class);
        retryMessages.stream().forEach(retryMessage -> {
            String exchange = retryMessage.getExchange();
            String routingKey = retryMessage.getRoutingKey();
            String body = retryMessage.getBody();
            String heads = retryMessage.getHeaders();
            MessageProperties messageProperties = new MessageProperties();

            String id = retryMessage.getMsgId();
            messageProperties.setMessageId(id);

            Map headers = JacksonUtil.json2BeanByZoneTime(heads, Map.class);
            if(!CollectionUtils.isEmpty(headers)){
                headers.forEach((k,v)->{
                    messageProperties.setHeader(k.toString(),v);
                });
            }
            Message message = new Message(body.getBytes(),messageProperties);


            amqpTemplate.send(exchange,routingKey,message);
        });
    }
}
